package com.my.hbase.orm;

import com.my.hbase.HBaseTemplate;
import com.my.hbase.demo.User;
import com.my.hbase.util.HBaseUtil;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;


/**
 * 基础操作类
 */
public class Model<M> {


    private static Logger logger = LoggerFactory.getLogger(Model.class);

    private Connection connection = HBaseTemplate.getConnection();

    /**
     * 新增
     */
    public void add(M obj) {

        String tableName = HBaseUtil.getTableName(obj);

        Put put = null;
        try {
            put = HBaseUtil.bean2Put(obj);
        } catch (Exception e) {
            logger.error("实体类转换put失败："+ e);
        }

        Table table = null;
        try {
            logger.info("##############hbase保存"+tableName+"开始##################");
            table = connection.getTable(TableName.valueOf(tableName));
            table.put(put);
            logger.info("##############hbase保存"+tableName+"结束["+put+"]##################");
        } catch (IOException e) {
            logger.error("HBase保存数据失败"+ e);
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                logger.error("HBase关闭连接失败"+ e);
            }
        }
    }

    /**
     * 删除
     */
    public void delete(M obj) {

        String rowkey = HBaseUtil.parseObjRowKey(obj);
        Delete delete = new Delete(Bytes.toBytes(rowkey));

        Table table = null;
        String tableName = HBaseUtil.getTableName(obj);
        try {
            logger.info("##############hbase删除，表" + tableName + "[" + rowkey + "]记录开始##################");
            table = connection.getTable(TableName.valueOf(tableName));
            table.delete(delete);
            logger.info("##############hbase删除，表" + tableName + "[" + rowkey + "]记录结束##################");
        } catch (IOException e) {
            logger.info("##############hbase删除，表" + tableName + "[" + rowkey + "]记录失败##################");
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                logger.error("HBase关闭连接失败" + e);
            }
        }
    }

    /**
     * 批量删除
     */
    public void batchDelete(String tableName, String[] rowkeys){
        List<Delete> deletes = new ArrayList<Delete>();

        for (String rowkey : rowkeys){
            Delete  delete = new Delete(Bytes.toBytes(rowkey));
            deletes.add(delete);
        }

        Table table = null;

        try {
            logger.info("##############hbase批量删除记录开始##################");
            table = connection.getTable(TableName.valueOf(tableName));
            table.delete(deletes);
            logger.info("##############hbase批量删除记录结束##################");
        } catch (IOException e) {
            logger.info("##############hbase批量删除记录失败##################");
        } finally {
            try {
                table.close();
            } catch (IOException e) {
                logger.error("HBase关闭连接失败" + e);
            }
        }
    }

    /**
     * 表格创建
     */
    public void createTable(M obj){
        String tableName = HBaseUtil.getTableName(obj);
        String[] families = HBaseUtil.getFamilies(obj);

        HBaseTemplate.createTable(tableName, families, false);
    }

    public M get(String tableName, String rowkey, Object obj){
        Get get = new Get(Bytes.toBytes(rowkey));

        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));

            Result result = table.get(get);

            if (!result.isEmpty()){
                try {
                    HBaseUtil.result2Bean(result, obj);
                } catch (Exception e) {
                    logger.error("HBase结果转换bean失败"+ e);
                    e.printStackTrace();
                }
            }
        } catch (IOException e) {
            logger.error("HBase打开连接失败"+ e);
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                logger.error("HBase关闭连接失败"+ e);
            }
        }

        return  (M)obj;
    }

    public Map<String, Object> get(String tableName, String rowkey){
        Get get = new Get(Bytes.toBytes(rowkey));

        Table table = null;
        try {
            table = connection.getTable(TableName.valueOf(tableName));

            Result result = table.get(get);

            if (!result.isEmpty()){
                List<Cell> cells = result.listCells();
                for (Cell cell : cells){

                }
            }
        } catch (IOException e) {
            logger.error("HBase打开连接失败"+ e);
        }finally {
            try {
                table.close();
            } catch (IOException e) {
                logger.error("HBase关闭连接失败"+ e);
            }
        }

        return null;
    }

    public static void  main(String arg[]){



        /*User user = new User();
        user.setRowkey("10001");
        user.setUserName("刘建康");
        user.setAge(23);
        user.setCreateTime(123456);*/

       Model<User> hBase = new Model<User>();

        /*
        //创建表
        hBase.createTable(user);
       */
        //添加记录
      //hBase.add(user);

        //删除记录
        //hBase.delete(user);

        //查询单条
        User user = hBase.get("user","10001", new User());

        System.out.println(user.getUserName());
        System.out.println(user.getAge());
        System.out.println(user.getCreateTime());

       /* User user = new User();

        Class cl = user.getClass();

        Field[] fields =   cl.getDeclaredFields();
        for(Field field : fields){

            System.out.println(field.getType().getName());

            if("[B".equals(field.getType().getName())){
                System.out.println("########");
            }
        }*/

    }
}
